/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.asiainfo.www;

import com.asiainfo.www.function.MapFileFlatMap;
import com.asiainfo.www.function.MapStatRichCoFlatMapFunction;
import com.asiainfo.www.function.StatFileFlatMap;
import com.asiainfo.www.function.RedisStatMapper;
import com.asiainfo.www.pojo.MapFile;
import com.asiainfo.www.pojo.StatFile;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Collector;
import scala.Tuple4;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class StreamingJob {

	public static void main(String[] args) throws Exception {
		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		//设置检查点时间
		env.enableCheckpointing(50000);
		//设置并行度
		env.setParallelism(10);

		//初始化kafka配置
		/**
		 *  接入上发信息
		 * */
		Properties properties = new Properties();
		//注意，程序是通过kafka连接zookeeper，然后通过zookeeper来找到kafka，zookeeper中配置的如果是hostname，本机也需要配置
		properties.setProperty("bootstrap.servers","192.168.61.131:9092,192.168.61.131:9093,192.168.61.131:9094");
		properties.setProperty("group.id","mapFile");
		DataStream<String> gprsUpStatFile = env
				.addSource(new FlinkKafkaConsumer010<String>("gprsUpStatFile",new SimpleStringSchema(),properties));
		//配置redis
		FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.61.131").build();
		//实例化RedisSink，并通过flink的addSink的方式将flink计算的结果插入到redis
		DataStream<StatFile> upStatFileDataStream = gprsUpStatFile.flatMap(new StatFileFlatMap());
		upStatFileDataStream.addSink(new RedisSink<StatFile>(conf,new RedisStatMapper()));
		upStatFileDataStream.print();
		/**
		 *  接入下发信息
		 * */
		DataStream<String> gprsDnStatFile = env
				.addSource(new FlinkKafkaConsumer010<String>("gprsDnStatFile",new SimpleStringSchema(),properties));
		DataStream<StatFile> dnStatFileDataStream = gprsDnStatFile.flatMap(new StatFileFlatMap());
		/**
		 *  接入map信息
		 * */
		DataStream<String> gprsMapStatFile = env
				.addSource(new FlinkKafkaConsumer010<String>("gprsMapFile",new SimpleStringSchema(),properties));
		DataStream< MapFile> mapFileDataStream = gprsMapStatFile.flatMap(new MapFileFlatMap());
		/**
		 *  进行key by
		 * */
		KeyedStream<StatFile,String> statFileStringKeyedStream = dnStatFileDataStream.keyBy(new KeySelector<StatFile, String>() {
			@Override
			public String getKey(StatFile statFile) throws Exception {
				return statFile.getDestFileName();
			}
		});

		KeyedStream<MapFile,String> mapFileStringKeyedStream = mapFileDataStream.keyBy(new KeySelector<MapFile, String>() {
			@Override
			public String getKey(MapFile mapFile) throws Exception {
				return mapFile.getDnName();
			}
		});
		//进行匹配
		ConnectedStreams<MapFile, StatFile> connectedStreams = mapFileStringKeyedStream.connect(statFileStringKeyedStream);
		SingleOutputStreamOperator<Tuple4<String, MapFile, StatFile, StatFile>> matchFileStreams = connectedStreams.flatMap(new MapStatRichCoFlatMapFunction());
		//进行选择数据分流
		SplitStream<Tuple4<String, MapFile, StatFile, StatFile>> matchFileSplitStreams = matchFileStreams.split(new OutputSelector<Tuple4<String, MapFile, StatFile, StatFile>>() {
			@Override
			public Iterable<String> select(Tuple4<String, MapFile, StatFile, StatFile> stringMapFileStatFileStatFileTuple4) {
				List<String> output = new ArrayList<>();
				output.add(stringMapFileStatFileStatFileTuple4._1());
				return output;
			}
		});
		//1.选择没有完成匹配的列
		DataStream<Tuple4<String, MapFile, StatFile, StatFile>> unmatchStreams = matchFileSplitStreams.select("UNMATCH");
		DataStream<String> unmatchStringStreams =  unmatchStreams.flatMap(new FlatMapFunction<Tuple4<String, MapFile, StatFile, StatFile>, String>() {
			@Override
			public void flatMap(Tuple4<String, MapFile, StatFile, StatFile> stringMapFileStatFileStatFileTuple4, Collector<String> collector) throws Exception {
				collector.collect(stringMapFileStatFileStatFileTuple4._2().output());
			}
		});

		unmatchStringStreams.print();
		unmatchStringStreams.addSink(new FlinkKafkaProducer010<String>("gprsUnmatchFile",new SimpleStringSchema(),properties));

		//2.选择匹配完成的列
		DataStream<Tuple4<String, MapFile, StatFile, StatFile>> matchStreams = matchFileSplitStreams.select("MATCH");
		DataStream<String> matchStringStreams =  matchStreams.flatMap(new FlatMapFunction<Tuple4<String, MapFile, StatFile, StatFile>, String>() {
			@Override
			public void flatMap(Tuple4<String, MapFile, StatFile, StatFile> stringMapFileStatFileStatFileTuple4, Collector<String> collector) throws Exception {
				collector.collect(stringMapFileStatFileStatFileTuple4._2().toString()+","+stringMapFileStatFileStatFileTuple4._3().toString()+","+stringMapFileStatFileStatFileTuple4._4().toString());
			}
		});
		matchStringStreams.print();
		matchStringStreams.addSink(new FlinkKafkaProducer010<String>("gprsMatchFile",new SimpleStringSchema(),properties));

		// execute program
		env.execute("UpStat Test.");
	}
}
